from absl import app, flags
import pandas as pd
import concurrent.futures
import datetime
from google.protobuf.json_format import MessageToDict

from coin.exchange.okex_futures.kr_rest.product_info import OkexFuturesProduct
from coin.util.queue.tools.kafka_archive import run_from_kafka_archive
from coin.util.queue.config import KafkaConfig
from coin.proto.coin_strategy_pb2 import StrategyLog
from coin.proto.coin_order_gateway_pb2 import OrderGatewayLog

FLAGS = flags.FLAGS

OKEX_REQUIREMENTS = {
    'BTC-USDT': [5, 20000],
    'ETH-USDT': [20, 10000],
    'EOS-USDT': [20, 10000],
    'LTC-USDT': [20, 10000],
    'BCHABC-USDT': [20, 10000],
    'XRP-USDT': [30, 7500],
    'OKB-USDT': [30, 15000],
    'ETC-USDT': [30, 7500],
    'ADA-USDT': [40, 7500],
    'NEO-USDT': [40, 7500],
    'TRX-USDT': [40, 4000],
    'BCHSV-USDT': [40, 4000],
    'DASH-USDT': [40, 7500],
    'QTUM-USDT': [40, 7500],
    'XLM-USDT': [40, 7500],
    'BTC-USDK': [40, 7500],
    'ETH-USDK': [40, 7500],
    'EOS-USDK': [40, 7500],
    'LTC-USDK': [40, 7500],
    'BTC-USD.THIS_WEEK': [5, 90000],
    'BTC-USD.NEXT_WEEK': [40, 30000],
    'BTC-USD.QUARTER': [5, 90000],
    'ETH-USD.THIS_WEEK': [5, 90000],
    'ETH-USD.NEXT_WEEK': [40, 30000],
    'ETH-USD.QUARTER': [5, 90000],
    'EOS-USD.THIS_WEEK': [20, 60000],
    'EOS-USD.NEXT_WEEK': [40, 30000],
    'EOS-USD.QUARTER': [20, 60000],
    'LTC-USD.THIS_WEEK': [20, 60000],
    'LTC-USD.NEXT_WEEK': [40, 30000],
    'LTC-USD.QUARTER': [20, 60000],
    'BCHABC-USD.THIS_WEEK': [20, 60000],
    'BCHABC-USD.NEXT_WEEK': [40, 30000],
    'BCHABC-USD.QUARTER': [20, 60000],
    'XRP-USD.THIS_WEEK': [40, 30000],
    'XRP-USD.NEXT_WEEK': [60, 15000],
    'XRP-USD.QUARTER': [40, 30000],
    'ETC-USD.THIS_WEEK': [40, 30000],
    'ETC-USD.NEXT_WEEK': [60, 15000],
    'ETC-USD.QUARTER': [40, 30000],
    'TRX-USD.THIS_WEEK': [60, 15000],
    'TRX-USD.NEXT_WEEK': [60, 15000],
    'TRX-USD.QUARTER': [60, 15000],
    'BCHSV-USD.THIS_WEEK': [60, 15000],
    'BCHSV-USD.NEXT_WEEK': [60, 15000],
    'BCHSV-USD.QUARTER': [60, 15000],
    'BTC-USD.PERPETUAL': [5, 120000],
    'EOS-USD.PERPETUAL': [20, 60000],
    'ETH-USD.PERPETUAL': [20, 60000],
    'LTC-USD.PERPETUAL': [20, 60000],
    'BCHABC-USD.PERPETUAL': [20, 60000],
    'ETC-USD.PERPETUAL': [30, 30000],
    'XRP-USD.PERPETUAL': [30, 30000],
    'TRX-USD.PERPETUAL': [50, 15000],
    'BCHSV-USD.PERPETUAL': [50, 15000],
}

HUOBI_REQUIREMENTS = {
    'ADA-USDT': [100, 30000],
    'BCHABC-USDT': [50, 3],
    'BCHABC-USDT': [50, 4],
    'BCHSV-USDT': [100, 7],
    'BTC-USDT': [50, 0.8],
    'EOS-BTC': [50, 300],
    'EOS-ETH': [50, 150],
    'EOS-USDT': [50, 300],
    'ETC-BTC': [100, 500],
    'ETC-USDT': [100, 150],
    'ETH-BTC': [50, 4],
    'ETH-USDT': [50, 15],
    'HT-USDT': [50, 350],
    'LTC-BTC': [50, 20],
    'LTC-USDT': [50, 50],
    'XLM-USDT': [100, 10000],
    'XMR-BTC': [100, 10],
    'XRP-BTC': [100, 8000],
    'XRP-USDT': [50, 8000],
    'BTC-HUSD': [50, 0.25],
}

BINANCE_REQUIREMENTS = {
    'BCHABC-USDT': [50, 3.5],
    'EOS-USDT': [50, 306],
    'LTC-USDT': [100, 15.6],
    'XRP-USDT': [70, 3900],
    'IOST-USDT': [200, 67900],
    'ADA-USDT': [100, 22000],
    'ONT-USDT': [150, 6940],
    'LINK-USDT': [100, 540],
    'TRX-USDT': [70, 64000],
    'ETH-USDT': [50, 5.9],
    'ETC-USDT': [50, 159],
    'XLM-USDT': [150, 7900]
}

REQUIREMENTS = {
    'Binance': BINANCE_REQUIREMENTS, 'Okex': OKEX_REQUIREMENTS, 'Huobi': HUOBI_REQUIREMENTS
}


def multithread_check_executor(func, param_list):
  results = []
  workers = len(param_list)
  if workers > 0:
    with concurrent.futures.ProcessPoolExecutor(max_workers=workers) as executor:
      future_list = [executor.submit(func, *param) for param in param_list]
    for future in future_list:
      results.append(future.result())
  return results


class OrderChecker(object):
  def __init__(self):
    self.order = {}

  def handle(self, partition, record):
    topic = record.topic
    pb = StrategyLog()
    pb.ParseFromString(record.value)
    if pb.type == StrategyLog.OG_LOG:
      og_log = pb.og_log
      if og_log.type == OrderGatewayLog.ORDER_EVENT:
        event = og_log.event
        orders = self.order.get(topic, [])
        orders.append(event)
        self.order.update({topic: orders})


def check_orders(topic,
                 orders,
                 start_ts,
                 end_ts,
                 market_type,
                 exchange,
                 precision=1e+9,
                 order_source='EXCHANGE'):
  if orders is None:
    return pd.DataFrame()
  if order_source == 'EXCHANGE':
    order_start_type = ['ORDER_ACCEPTED']
    order_end_type = ['ORDER_FILLED', 'CANCEL_CONFIRMED']
  elif order_source == 'USER':
    order_start_type = ['ORDER_SUBMITTED', 'ORDER_ACCEPTED']
    order_end_type = ['CANCEL_SUBMITTED', 'ORDER_FILLED', 'CANCEL_CONFIRMED']
  else:
    raise ValueError("Invalid order source", order_source)
  df_list = []
  time_range = range(int(start_ts / precision), int(end_ts / precision))
  total_score = []
  for order in orders:
    df_list.append(MessageToDict(order))
  order_df = pd.DataFrame(df_list)
  order_df.sort_values('eventTime', inplace=True)
  order_df = order_df[order_df['orderEventSource'] == order_source]
  for symbol in set(order_df['symbol'].tolist()):
    score = pd.DataFrame(index=time_range,
                         columns=['buy_price', 'buy_qty', 'sell_price', 'sell_qty', 'end_type'])
    symbol_order = order_df[order_df['symbol'] == symbol]
    order_start = symbol_order[symbol_order['type'].isin(order_start_type)]
    order_end = symbol_order[symbol_order['type'].isin(order_end_type)]
    for index in order_start.index:
      order_id = order_start.loc[index, 'externalOrderId']
      if order_id in order_end['externalOrderId'].tolist():
        start_time = order_start.loc[index, 'eventTime']
        end_time = order_end['eventTime'][order_end['externalOrderId'] == order_id].iloc[0]
        end_type = order_end['type'][order_end['externalOrderId'] == order_id].iloc[0]
        start_time = float(start_time)
        end_time = float(end_time)
        if end_time - start_time > 1 * precision:
          price = order_start.loc[index, 'orderPrice']
          qty = order_start.loc[index, 'orderQty']
          side = order_start.loc[index, 'orderSide']
          end_time = int(end_time / precision)
          start_time = int(start_time / precision)
          if 'BUY' in side:
            score.loc[start_time:end_time, ['buy_price', 'buy_qty']] = [price, qty]
          elif 'SELL' in side:
            score.loc[start_time:end_time, ['sell_price', 'sell_qty']] = [price, qty]
          score.loc[end_time, 'end_type'] = end_type
    if market_type == 'Futures' or market_type == 'Swap':
      product = OkexFuturesProduct.FromStr(symbol)
      symbol = product.subscription_symbol
    order_rate = len(score[['buy_price', 'sell_price']].dropna()) / len(score)
    one_side_rate = len(score[['buy_price', 'sell_price']].dropna(how='all')) / len(score)
    score['spread'] = (score['sell_price']-score['buy_price'])\
                      /(score['sell_price']+score['buy_price'])*2*10000
    if market_type == 'Spot':
      if exchange in ['Okex', 'Binance']:
        score['depth'] = (pd.concat(
            [score['sell_qty'] * score['sell_price'], score['buy_qty'] * score['buy_price']],
            axis=1)).dropna().min(axis=1)
      elif exchange in ['Huobi']:
        score['depth'] = (pd.concat([score['sell_qty'], score['buy_qty']],
                                    axis=1)).dropna().min(axis=1)
    else:
      multiplier = 100 if 'BTC' in symbol else 10
      score['depth'] = (pd.concat([multiplier * score['sell_qty'], multiplier * score['buy_qty']],
                                  axis=1)).dropna().min(axis=1)
    score['feed_spread'] = score['spread'] < REQUIREMENTS[exchange][symbol][0]
    score['feed_depth'] = score['depth'] > REQUIREMENTS[exchange][symbol][1]
    spread_score, depth_score = score['feed_spread'].sum()/len(score), \
                                score['feed_depth'].sum()/len(score)
    total_score.append([symbol, spread_score, depth_score, order_rate, one_side_rate])
  total_score = pd.DataFrame(total_score,
                             columns=['symbol', 'spread', 'depth', 'order_rate', 'one_side_rate'])
  print("topic: ", topic)
  print(total_score.to_string())
  return total_score


def report_happy_score(start_time, end_time, topics, market_type, exchange):
  kafka_config = KafkaConfig.from_cmd_config(FLAGS.kafka_config_filename)
  handler = OrderChecker()
  run_from_kafka_archive(topics=topics,
                         kafka_config=kafka_config,
                         timestamp_from=start_time,
                         timestamp_to=end_time,
                         callback=handler.handle)
  print("start: ", datetime.datetime.utcfromtimestamp(start_time / 1e+9))
  print("end: ", datetime.datetime.utcfromtimestamp(end_time / 1e+9))
  params_list = []
  for topic in topics:
    params = [topic, handler.order.get(topic), start_time, end_time, market_type, exchange]
    params_list.append(params)
    if not FLAGS.multiprocess:
      score = check_orders(topic=topic,
                           orders=handler.order.get(topic),
                           start_ts=start_time,
                           end_ts=end_time,
                           market_type=market_type,
                           exchange=exchange)
  if FLAGS.multiprocess:
    multithread_check_executor(check_orders, params_list)


def main(_):
  Okex_Spot_topics = [
      'STRAT_strategy-14.aliyun-cn-hongkong_dmm_okex_binance_usdt_1_1',
      'STRAT_strategy-14.aliyun-cn-hongkong_dmm_okex_binance_usdt_2_1',
      'STRAT_strategy-14.aliyun-cn-hongkong_dmm_okex_binance_usdt_3_1',
      'STRAT_strategy-16.aliyun-cn-hongkong_dmm_okex_binance_usdt_1_2',
      'STRAT_strategy-16.aliyun-cn-hongkong_dmm_okex_binance_usdt_2_2',
      'STRAT_strategy-16.aliyun-cn-hongkong_dmm_okex_binance_usdt_3_2',
  ]

  Okex_Futures_topics = [
      'STRAT_strategy-18.aliyun-cn-hongkong_dmm_okex_this_week_usd_1_1',
      'STRAT_strategy-18.aliyun-cn-hongkong_dmm_okex_this_week_usd_2_1',
      'STRAT_strategy-33.aliyun-cn-hongkong_dmm_okex_next_week_usd_1_1',
      'STRAT_strategy-33.aliyun-cn-hongkong_dmm_okex_quarter_usd_1_1',
      'STRAT_strategy-20.aliyun-cn-hongkong_dmm_okex_this_week_usd_1_2',
      'STRAT_strategy-20.aliyun-cn-hongkong_dmm_okex_this_week_usd_2_2',
      'STRAT_strategy-34.aliyun-cn-hongkong_dmm_okex_next_week_usd_1_2',
      'STRAT_strategy-34.aliyun-cn-hongkong_dmm_okex_quarter_usd_1_2',
  ]

  Okex_Swap_topics = [
      'STRAT_strategy-37.aliyun-cn-hongkong_dmm_okex_swap_usd_1_1',
      'STRAT_strategy-37.aliyun-cn-hongkong_dmm_okex_swap_usd_2_1',
      'STRAT_strategy-37.aliyun-cn-hongkong_dmm_okex_swap_usd_1_2',
      'STRAT_strategy-37.aliyun-cn-hongkong_dmm_okex_swap_usd_2_2'
  ]

  Binance_Spot_topics = ['STRAT_strategy-20.ap-northeast-1_dmm_binance_usdt_1']

  Huobi_Spot_topics = [
      'STRAT_strategy-09.ap-northeast-1.huobi_dmm_huobi_binance_usdt_2',
      'STRAT_strategy-09.ap-northeast-1.huobi_dmm_huobi_binance_btc_1'
  ]
  start_time = FLAGS.start_time
  end_time = FLAGS.end_time
  trading_date = FLAGS.trading_date
  if trading_date:
    for date in trading_date.split(','):
      date = datetime.datetime.strptime(date, '%Y-%m-%d')
      start_time = int(datetime.datetime.combine(date, datetime.time.min).timestamp() * 1e+9)
      end_time = int(datetime.datetime.combine(date, datetime.time.max).timestamp() * 1e+9)
  elif start_time and end_time:
    start_time = int(datetime.datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S').timestamp() * 1e+9)
    end_time = int(datetime.datetime.strptime(end_time, '%Y-%m-%d %H:%M:%S').timestamp() * 1e+9)
  else:
    date = datetime.datetime.utcnow().date() - datetime.timedelta(days=1)
    start_time = int(datetime.datetime.combine(date, datetime.time.min).timestamp() * 1e+9)
    end_time = int(datetime.datetime.combine(date, datetime.time.max).timestamp() * 1e+9)
  market_type = FLAGS.market_type
  exchange = FLAGS.exchange
  assert market_type in ['Spot', 'Futures', 'Swap']
  assert exchange in ['Okex', 'Binance', 'Huobi']
  topics = locals()['%s_%s_topics' % (exchange, market_type)]
  report_happy_score(start_time, end_time, topics, market_type, exchange)


if __name__ == '__main__':
  flags.DEFINE_string('kafka_config_filename',
                      '../../coin_deploy/support_monitor/config/kafka_config.json',
                      'kafka config')
  flags.DEFINE_string('strategy_config_filename',
                      '../../coin_deploy/support_monitor/config/strat_config/strat_all.json',
                      'strategy_config_filename')
  flags.DEFINE_string('start_time', None, 'start time')
  flags.DEFINE_string('end_time', None, 'end time')
  flags.DEFINE_string('trading_date', None, 'trading date')
  flags.DEFINE_string('market_type', None, 'market type')
  flags.DEFINE_string('exchange', None, 'exchange')
  flags.DEFINE_boolean('multiprocess', False, '')
  app.run(main)
